package simpledb.storage;

import simpledb.common.Database;
import simpledb.common.DbException;
import simpledb.common.Permissions;
import simpledb.transaction.TransactionAbortedException;
import simpledb.transaction.TransactionId;

import java.io.File;
import java.io.FileNotFoundException;
import java.io.IOException;
import java.io.RandomAccessFile;
import java.util.*;

/**
 * HeapFile is an implementation of a DbFile that stores a collection of tuples
 * in no particular order. Tuples are stored on pages, each of which is a fixed
 * size, and the file is simply a collection of those pages. HeapFile works
 * closely with HeapPage. The format of HeapPages is described in the HeapPage
 * constructor.
 * 
 * @see HeapPage#HeapPage
 * @author Sam Madden
 */
public class HeapFile implements DbFile {
    private File dbFile;
    private TupleDesc td;
    /**
     * Constructs a heap file backed by the specified file.
     * 
     * @param f
     *            the file that stores the on-disk backing store for this heap
     *            file.
     */
    public HeapFile(File f, TupleDesc td) {
        // some code goes here
        dbFile=f;
        this.td=td;
    }

    /**
     * Returns the File backing this HeapFile on disk.
     * 
     * @return the File backing this HeapFile on disk.
     */
    public File getFile() {
        // some code goes here
        return dbFile;
    }

    /**
     * Returns an ID uniquely identifying this HeapFile. Implementation note:
     * you will need to generate this tableid somewhere to ensure that each
     * HeapFile has a "unique id," and that you always return the same value for
     * a particular HeapFile. We suggest hashing the absolute file name of the
     * file underlying the heapfile, i.e. f.getAbsoluteFile().hashCode().
     * 
     * @return an ID uniquely identifying this HeapFile.
     */
    public int getId() {
        // some code goes here
        return dbFile.getAbsoluteFile().hashCode();
    }

    /**
     * Returns the TupleDesc of the table stored in this DbFile.
     * 
     * @return TupleDesc of this DbFile.
     */
    public TupleDesc getTupleDesc() {
        // some code goes here
        return td;
    }

    // see DbFile.java for javadocs
    public Page readPage(PageId pid){
        // some code goes here
        RandomAccessFile randomAccessFile=null;
        byte[]buffer=new byte[BufferPool.getPageSize()];
        try {
            randomAccessFile=new RandomAccessFile(dbFile,"r");
            long len=dbFile.length();
            //先确保长度没有问题。
            if((long) pid.getPageNumber() *BufferPool.getPageSize()>=len){
                return null;
            }
            //接下来确保入口。
            int startPosition=pid.getPageNumber()*BufferPool.getPageSize();
            randomAccessFile.seek(startPosition);
            int offset= (int) Math.min(BufferPool.getPageSize(),len - BufferPool.getPageSize() * pid.getPageNumber());
            //进行正确的读操作。
            //接下来去读,我靠，之前出错是因为第二个参数是数组的起始地方位置，永远是0.
            randomAccessFile.read(buffer,0,offset);

        } catch (FileNotFoundException e) {
            e.printStackTrace();
        } catch (IOException e) {
            e.printStackTrace();
        }
        Page p = null;
        try {
            p = new HeapPage(new HeapPageId(pid.getTableId(),pid.getPageNumber()),buffer);
        } catch (IOException e) {
            e.printStackTrace();
        }
        return p;
    }

    // see DbFile.java for javadocs
    public void writePage(Page page) throws IOException {
        RandomAccessFile raf = new RandomAccessFile(this.dbFile, "rw");
        long offset = page.getId().getPageNumber() * BufferPool.getPageSize();
        raf.seek(offset);
        raf.write(page.getPageData(),0,BufferPool.getPageSize());
    }

    /**
     * Returns the number of pages in this HeapFile.
     */
    public int numPages() {
        // some code goes here
        return (int) (dbFile.length()/BufferPool.getPageSize());
    }

    // see DbFile.java for javadocs
    public List<Page> insertTuple(TransactionId tid, Tuple t)
            throws DbException, IOException, TransactionAbortedException {
        ArrayList<Page> list = new ArrayList<>();
        BufferPool pool = Database.getBufferPool();
        int tableid = getId();
        for (int i = 0; i < numPages(); ++i) {
            HeapPage page = (HeapPage)pool.getPage(tid, new HeapPageId(tableid, i), Permissions.READ_WRITE);
            if (page.getNumEmptySlots() > 0) {
                page.insertTuple(t);
                page.markDirty(true, tid);
                list.add(page);
                return list;
            }
        }
        HeapPage page = new HeapPage(new HeapPageId(tableid, numPages()), HeapPage.createEmptyPageData());
        page.insertTuple(t);
        writePage(page);
        return list;
    }

    // see DbFile.java for javadocs
    public ArrayList<Page> deleteTuple(TransactionId tid, Tuple t) throws DbException,
            TransactionAbortedException, IOException {
        HeapPageId pid = (HeapPageId) t.getRecordId().getPageId();
        if (pid.getTableId() != getId()) {
            throw new DbException("tuple not a member of this table");
        }
        HeapPage page = (HeapPage) Database.getBufferPool().getPage(tid, pid, Permissions.READ_WRITE);
        page.deleteTuple(t);
        page.markDirty(true, tid);
        return new ArrayList<>(Arrays.asList(page));
    }

    // see DbFile.java for javadocs
    public DbFileIterator iterator(TransactionId tid) {

        // some code goes here
        return new DbFileIterator() {
            private Iterator<Tuple>iterator;
            private int pageIndex;
            @Override
            public void open() throws DbException, TransactionAbortedException{
               HeapPage page=getPageByBufferPool(0);
                pageIndex=0;
                iterator= page.iterator();
            }

            @Override
            public boolean hasNext() throws DbException, TransactionAbortedException{
                if(iterator==null)
                  return false;
                //淦，java为毛后置这玩意能过啊？前面false后面不应该去访问啊？
                if(iterator.hasNext()){
                    return true;
                }
                else if(pageIndex+1<numPages()){
                   pageIndex++;
                    HeapPage page= null;
                    page = getPageByBufferPool(pageIndex);
                    iterator= page.iterator();
                   return hasNext();
                }
                else
                return false;
            }

            @Override
            public Tuple next() throws DbException, TransactionAbortedException, NoSuchElementException{
                if(hasNext()){
                    return iterator.next();
                }
                throw new NoSuchElementException("no next element");
            }

            @Override
            public void rewind() throws DbException, TransactionAbortedException{
                open();
            }

            @Override
            public void close() {
                iterator = null;
                pageIndex = 0;
            }
            public HeapPage getPageByBufferPool(int pgNo) throws TransactionAbortedException, DbException{
                HeapPageId heapPageId=new HeapPageId(getId(),pgNo);
                HeapPage page=(HeapPage) Database.getBufferPool().getPage(tid,heapPageId,Permissions.READ_ONLY);
                return page;
            }
        };
    }

}

